package com.shujia.flink.table

import org.apache.flink.table.api.{EnvironmentSettings, TableEnvironment}

object Demo3BatchSQL {
  def main(args: Array[String]): Unit = {
    //创建环境设置对象
    val settings: EnvironmentSettings = EnvironmentSettings
      .newInstance()
      //.inStreamingMode() //流处理模式，输出连续结果
      .inBatchMode() //批处理模式，输出最终结果
      .build()

    //创建flink sql环境
    val tEnv: TableEnvironment = TableEnvironment.create(settings)

    /**
     * 2、基于文件创建表  -- 有界流   -- source表
     */
    tEnv.executeSql(
      """
        |CREATE TABLE students (
        |  id STRING,
        |  name STRING,
        |  age INT,
        |  sex STRING,
        |  clazz STRING
        |)  WITH (
        |  'connector' = 'filesystem',           -- 必选：指定连接器类型
        |  'path' = 'data/students.txt',  -- 必选：指定路径
        |  'format' = 'csv'                     -- 必选：文件系统连接器指定 format
        |)
        |""".stripMargin)


    /**
     * 2、创建sink表，保存结果
     */

    tEnv.executeSql(
      """
        |CREATE TABLE clazz_num (
        |  clazz STRING,
        |  num BIGINT
        |)  WITH (
        |  'connector' = 'filesystem',           -- 必选：指定连接器类型
        |  'path' = 'data/clazz_num',  -- 必选：指定路径
        |  'format' = 'csv'                     -- 必选：文件系统连接器指定 format
        |)
        |""".stripMargin)

    /**
     * 统计班级的人数
     */

    tEnv.executeSql(
      """
        |insert into clazz_num
        |select
        |clazz,count(1) as num
        |from
        |students
        |group by
        |clazz
        |""".stripMargin)

  }

}
